package com.mjp.zookeeper.watcher;

import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class WatchCallBackLock implements Watcher,
        AsyncCallback.StringCallback,
        AsyncCallback.ChildrenCallback,
        AsyncCallback.StatCallback {

    private ZooKeeper zooKeeper;

    private String threadName;

    private CountDownLatch cc = new CountDownLatch(1);

    private String nodeName;

    public WatchCallBackLock(ZooKeeper zooKeeper, String threadName) {
        this.zooKeeper = zooKeeper;
        this.threadName = threadName;
    }

    public void tryLock() {
        try {
            zooKeeper.create("/fbsLock",
                    threadName.getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL,
                    this, "abc");
            cc.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void unLock() {
        try {
            System.out.println(threadName + " 干活结束，释放节点 ");
            zooKeeper.delete("/" + nodeName, -1);
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * watch 实现
     *
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        switch (event.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                //节点删除的时候,
                zooKeeper.getChildren("/", false, this, "abc");
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
            case PersistentWatchRemoved:
                break;
        }
    }

    /**
     * stringCallback
     *
     * @param rc
     * @param path
     * @param ctx
     * @param name
     */
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        System.out.println(threadName + " create node " + name);

        if (!StringUtils.isEmpty(name)) {
            this.nodeName = name.substring(1);
            zooKeeper.getChildren("/", false, this, "abc");
        }
    }

    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children) {
        if (!children.isEmpty()) {
            Collections.sort(children);
            int index = children.indexOf(nodeName);
            if (index == 0) {
                //如果index 为0 表示当前在第一位 可以拿到锁，继续执行
                cc.countDown();
            } else {
                //如果index 不为0 就观察前一个节点
                zooKeeper.exists("/" + children.get(index - 1), this, this, "abc");

            }
        }
    }

    /**
     * statCallback
     *
     * @param rc
     * @param path
     * @param ctx
     * @param stat
     */
    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        if (stat == null) {
            cc.countDown();
        }
    }
}
